package com.vodafone.mapreduce.map;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.log4j.Logger;

import com.vodafone.mapreduce.MapReduce;
import com.vodafone.mapreduce.dto.common.KeyValuePair;
import com.vodafone.mapreduce.dto.function.CombineInput;

/**
 * The map function maps one (key, value) pair (k, v) to a set of (key, value) pairs (li, wi).
 * 
 * @param <K>
 * @param <V>
 * @param <L>
 * @param <W>
 */
public abstract class MapFunction<K, V, L, W> {

    public static Logger log = MapReduce.LOG;

    protected K k;
    protected V v;
    protected final List<KeyValuePair<L, W>> intermediate = new ArrayList<KeyValuePair<L, W>>();
    protected final AtomicBoolean ready = new AtomicBoolean(false);

    public void initialize(K k, V v) {
        ready.set(false);
        intermediate.clear();
        this.k = k;
        this.v = v;
    }

    public void map() {
        doMap(k, v);
        ready.set(true);
    }

    protected abstract void doMap(K k, V v);

    protected void emitIntermediate(L l, W w) {
        intermediate.add(new KeyValuePair<L, W>(l, w));
    }

    public CombineInput<K, V, L, W> getMapResult() {
        if (ready.get()) {
            return new CombineInput<K, V, L, W>(k, v, intermediate);
        } else {
            return null;
        }
    }

}
